package com.pai4j.connect.messagequeue.consumer.redis;

import com.pai4j.connect.messagequeue.consumer.service.InteractionNotificationConsumeService;
import com.pai4j.connect.messagequeue.util.RedisMessageUtil;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.StringUtils;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.data.redis.connection.Message;
import org.springframework.data.redis.connection.MessageListener;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.stereotype.Component;

import java.util.concurrent.ArrayBlockingQueue;

/**
 *  互动通知消息监听
 *
 * @author: CYM-pai
 * @date: 2025/07/28 23:51
 **/

@Slf4j
@Component
public class InteractionNotificationRedisQueueListener implements MessageListener {

    @Autowired
    private RedisTemplate redisTemplate;

    @Autowired
    private InteractionNotificationConsumeService interactionNotificationConsumeService;

    private static final int TASK_QUEUE_SIZE = 2000;

    /**
     * 阻塞队列入队、消费异常，重试次数
     */
    private static final int RETRY_TIMES = 3;

    private static ArrayBlockingQueue<String> taskQueue = new ArrayBlockingQueue<>(TASK_QUEUE_SIZE);

    @Override
    public void onMessage(Message message, byte[] pattern) {
        // 使用工具类安全地反序列化频道和消息
        String channel = RedisMessageUtil.safeDeserializeChannel(message, redisTemplate);
        String messageStr = RedisMessageUtil.safeDeserializeMessage(message, redisTemplate);
        
        if (messageStr == null) {
            log.warn("InteractionNotificationRedisQueueListener.onMessage =====> 接收到无效消息，已跳过处理. channel:{}", channel);
            return;
        }
        
        log.info("InteractionNotificationRedisQueueListener.onMessage =====> channel:{} messageStr:{}", channel, messageStr);
        // 加入任务队列（遇到通知大流量场景，缓解consumer压力）
        this.putTaskQueue(messageStr);
    }

    public void consumeTask() {
        while (true) {
            String message = "";
            try {
                message = taskQueue.take();
                if (StringUtils.isNotBlank(message)) {
                    interactionNotificationConsumeService.consume(message);
                }
            } catch (InterruptedException e) {
                log.error("互动通知消息队列消费异常！message:{}", message, e);
                this.putTaskQueue(message);
            } catch (Exception e) {
                log.error("互动通知消息队列消费异常！message:{}", message, e);
            }
        }
    }

    /**
     * 重新加入队列
     * @param message
     */
    private void putTaskQueue(String message) {
        int retryTimes = 0;
        while (retryTimes <= RETRY_TIMES) {
            try {
                taskQueue.put(message);
                return;
            } catch (InterruptedException e) {
                log.error("互动通知消息加入队列异常！message:{}", message, e);
                retryTimes ++;
            }
        }
    }

}
